﻿using Common.Extension;
using Common.Json;
using Common.MQ;
using Newtonsoft.Json.Linq;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading.Tasks;
using System.Transactions;

namespace Common.RabbitMQ
{
    /// <summary>
    /// RabbitMQ消费者
    /// </summary>
    public class RabbitConsumer : MQConsumerBase
    {

        private const ushort DefaultQos = 1;
        private readonly RabbitConfig connector;
        private readonly QueueSpec queue;
        /// <summary>
        /// RabbitMQ消费者
        /// </summary>
        /// <param name="connector"></param>
        /// <param name="queue"></param>
        /// <param name="handlers"></param>
        public RabbitConsumer(RabbitConfig connector, QueueSpec queue, params IMessageHandler[] handlers) : base(handlers?.ToList())
        {
            this.connector = connector;
            this.queue = queue;
        }
        /// <summary>
        /// 
        /// </summary>
        protected override void ConnectToListen()
        {
            var factory = new ConnectionFactory();
            factory.HostName = this.connector.Host;
            factory.Port = this.connector.Port;
            factory.UserName = this.connector.UserName;
            factory.Password = this.connector.Password;
            factory.AutomaticRecoveryEnabled = true;
            factory.TopologyRecoveryEnabled = true;
            factory.NetworkRecoveryInterval = TimeSpan.FromSeconds(10);
            if (!string.IsNullOrEmpty(this.queue.VirtualHost))
            {
                factory.VirtualHost = this.queue.VirtualHost;
            }
            var connection = factory.CreateConnection();
            var channel = connection.CreateModel();
            
            string exchangeType = this.queue.ExchangeType.GetStringValue();
            var exchangeProperties = new Dictionary<string, object> { { "x-dead-letter-exchange", queue.Exchange } };
            var normalProperties = new Dictionary<string, object> { { "x-dead-letter-exchange", $"{queue.Exchange}-DLX" } };
            //声明重试交换机，采用topic模式
            channel.ExchangeDeclare($"{queue.Exchange}-retry", exchangeType, true, false, null);
            //channel.ExchangeDeclare(RETRY_EXCHANGE, ExchangeType.Direct, true, false, null);
            ushort Qos = this.connector.Qos > 0 ? this.connector.Qos : DefaultQos;
            channel.BasicQos(0, Qos, false);
            channel.QueueDeclare(this.queue.Queue, true, false, false, normalProperties);
            channel.QueueDeclare($"{queue.Queue}-retry", true, false, false, exchangeProperties);

            channel.QueueBind($"{queue.Queue}-retry", $"{queue.Exchange}-retry", this.queue.RoutingKey, null);
            
            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += OnMessageReceived;
            channel.BasicConsume(this.queue.Queue, false, consumer);
        }


        private void OnMessageReceived(object sender, BasicDeliverEventArgs e)
        {
            var msg = Encoding.UTF8.GetString(e.Body.ToArray());
            string typeString = null;
            try
            {
                JObject jsonObject = JObject.Parse(msg);
                typeString = jsonObject.GetString("_TypeString");
            }
            catch
            {
                Console.WriteLine($"get message type failed, content: {msg}");
            }
            var consumer = sender as EventingBasicConsumer;
            //异步处理消息，（不能将参数[e]传入方法，handle方法是异步的，此方法return之后[e]随时可能被重新赋值，导致丢失消息数据）
            var handlers = GetMatchHandlers(typeString);
            var recvMsgInfo = new RecvMsgInfo
            {
                Sender = consumer.Model,
                DeliveryTag = e.DeliveryTag,
                Properties = e.BasicProperties.CopyTo(consumer.Model.CreateBasicProperties()),
                Body = e.Body.ToArray(),//通过字节数组的隐式转换获得新的ReadOnlyMemory<byte>对象，避免因异步导致的e.Body被回收，无法正常处理消息。
                Exchange = e.Exchange,
                RoutingKey = e.RoutingKey,
                Data = msg
            };
            HandleAsync(handlers, recvMsgInfo);
        }
        private Task HandleAsync(List<IMessageHandler> handlers, RecvMsgInfo msg)
        {
            return Task.Run(() =>
            {
                bool res = false;
                try
                {
                    if (handlers == null || !handlers.Any())
                    {
                        return;
                    }
                    res = handlers.Count == 1 ? ExecuteSingleHandler(handlers[0], msg.Data) : ExecuteMultiHandler(handlers, msg.Data);
                }
                catch
                {
                    res = false;
                }
                finally
                {
                    #region 处理消息的确认或拒绝
                    //如果没有拒绝或者确认消息，消费者端会被阻塞，造成无法再收到消息
                    if (res)
                    {
                        msg.Sender.BasicAck(msg.DeliveryTag, false);
                    }
                    else
                    {
                        var header = (msg.Properties.Headers?["x-death"] as List<object>)?.FirstOrDefault() as Dictionary<string, object>;
                        long deathCount = header != null && header.TryGetValue("count", out object temp) ? (long)temp : 0;

                        if (deathCount >= connector.RetryTimes)
                        {
                            //超过重试次数，发送到死信队列
                            msg.Sender.BasicReject(msg.DeliveryTag, false);
                        }
                        else
                        {
                            //先确认消息
                            msg.Sender.BasicAck(msg.DeliveryTag, false);
                            //再发送到重试队列，继续重试
                            msg.Properties.Expiration = ((long)connector.RetryInterval.TotalMilliseconds).ToString();
                            //var props = msg.Sender.CreateBasicProperties();
                            //props.Expiration = ((long)connector.RetryInterval.TotalMilliseconds).ToString();
                            //props.Headers = msg.Properties.Headers;
                            //props.DeliveryMode = msg.Properties.DeliveryMode;
                            //props.Type = msg.Properties.Type;
                            var address = new PublicationAddress(queue.ExchangeType.GetStringValue(), $"{queue.Exchange}-retry", queue.RoutingKey);
                            msg.Sender.BasicPublish(address, msg.Properties, msg.Body);

                        }
                    }
                    #endregion
                }
            });
        }
        private bool ExecuteSingleHandler(IMessageHandler handler, string data)
        {
            return handler.Handle(data);
        }
        private bool ExecuteMultiHandler(IEnumerable<IMessageHandler> handlers, string data)
        {
            using (TransactionScope ts = new TransactionScope())
            {
                foreach (var h in handlers)
                {
                    if (!h.Handle(data))
                    {
                        return false;
                    }
                }
                ts.Complete();
                return true;
            }
        }
    }
#pragma warning disable CS1591
    public class RecvMsgInfo
    {
        public ulong DeliveryTag { get; set; }
        public IBasicProperties Properties { get; set; }
        public ReadOnlyMemory<byte> Body { get; set; }
        public IModel Sender { get; set; }
        public string Exchange { get; set; }
        public string RoutingKey { get; set; }
        public string Data { get; set; }
    }
#pragma warning restore CS1591
}
